package com.udf.hive.udf.extract;

import java.util.logging.Logger;

import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.Text;

/**
 * 数据提取-单分隔符
 * 
 * @作者 tao zhou
 * @version [版本号, 2020年7月18日]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class StartEndExtractUDF extends UDF
{
    private static Logger logger = Logger.getLogger("StartEndExtractUDF");
    
    /**
     * 数据处理逻辑
     * 
     * @param inputField
     *            输入字段
     * @param splitStartStr
     *            开始分隔符
     * @param splitStartStrIndex
     *            开始分隔符位点
     * @param isRetainSplitStartStr
     *            是否保留开始分隔符 0: 否 ,1:是
     * @param splitEndStr
     *            结束分隔符
     * @param splitStartEndIndex
     *            结束分隔符位点
     * @param isRetainSplitEndStr
     *            是否保留结束分隔符0: 否 ,1:是
     * @return
     * @exception/throws [违例类型] [违例说明]
     * @see [类、类#方法、类#成员]
     */
    public Text evaluate(Text inputField, Text splitStartStr, Text splitStartStrIndex, Text isRetainSplitStartStr,
            Text splitEndStr, Text splitStartEndIndex, Text isRetainSplitEndStr) {
        if (inputField == null) {
            return null;
        }
        String inputFieldStr = String.valueOf(inputField).trim();
        String splitStart = String.valueOf(splitStartStr).trim();
        String splitStartStrIndexStr = String.valueOf(splitStartStrIndex).trim();
        String isRetainSplitStart = String.valueOf(isRetainSplitStartStr).trim();
        String splitEnd = String.valueOf(splitEndStr).trim();
        String splitStartEndIndexStr = String.valueOf(splitStartEndIndex).trim();
        String isRetainSplitEnd = String.valueOf(isRetainSplitEndStr).trim();
        Integer splitStartStrIndexInt = Integer.parseInt(splitStartStrIndexStr);
        Integer isRetainSplitStartInt = Integer.parseInt(isRetainSplitStart);
        Integer splitStartEndIndexInt = Integer.parseInt(splitStartEndIndexStr);
        Integer isRetainSplitEndInt = Integer.parseInt(isRetainSplitEnd);
        String result = null;
        // 目前只对整型做转换
        try {
            result = startEndExtract(inputFieldStr, splitStart, splitStartStrIndexInt, isRetainSplitStartInt, splitEnd,
                    splitStartEndIndexInt, isRetainSplitEndInt);
        }
        catch (Exception e) {
            e.printStackTrace();
            result = inputFieldStr;
        }

        return new Text(result);
    }

    public String startEndExtract(String inputFieldStr, String splitStart, Integer splitStartStrIndexInt,
            Integer isRetainSplitStartInt, String splitEnd, Integer splitStartEndIndexInt,
            Integer isRetainSplitEndInt) {
        String result;
        int index = 0;
        int startRegexIndex = 0;

        for (int i = 1; i <= splitStartStrIndexInt; i++) {
            startRegexIndex = inputFieldStr.indexOf(splitStart, (1 == i ? index : index + splitStart.length()));
            index = startRegexIndex;
        }
        index = 0;
        int endRegexIndex = 0;
        for (int i = 1; i <= splitStartEndIndexInt; i++) {
            endRegexIndex = inputFieldStr.indexOf(splitEnd, (1 == i ? index : index + splitEnd.length()));
            index = endRegexIndex;
        }
        result = inputFieldStr.substring(startRegexIndex + (1 == isRetainSplitStartInt ? 0 : splitStart.length()),
                endRegexIndex + (1 == isRetainSplitEndInt ? splitEnd.length() : 0));
        return result;
    }

    public static void main(String[] args) {
        logger.info(String.valueOf(new StartEndExtractUDF().evaluate(new Text("[江苏省][苏州市]"), new Text("["), new Text("1"),
                new Text("1"), new Text("]"), new Text("1"), new Text("1"))));
        logger.info(String.valueOf(new StartEndExtractUDF().evaluate(new Text("[江苏省][苏州市]"), new Text("["), new Text("1"),
                new Text("1"), new Text("]"), new Text("1"), new Text("0"))));
        logger.info(String.valueOf(new StartEndExtractUDF().evaluate(new Text("[江苏省][苏州市]"), new Text("["), new Text("1"),
                new Text("0"), new Text("]"), new Text("1"), new Text("0"))));
        logger.info(String.valueOf(new StartEndExtractUDF().evaluate(new Text("[江苏省][苏州市]"), new Text("["), new Text("2"),
                new Text("1"), new Text("]"), new Text("2"), new Text("1"))));
        logger.info(String.valueOf(new StartEndExtractUDF().evaluate(new Text("[江苏省][苏州市]"), new Text("["), new Text("2"),
                new Text("1"), new Text("]"), new Text("2"), new Text("0"))));
        logger.info(String.valueOf(new StartEndExtractUDF().evaluate(new Text("[江苏省][苏州市]"), new Text("["), new Text("2"),
                new Text("0"), new Text("]"), new Text("2"), new Text("0"))));
    }
}
